package com.central.streaming

import java.sql.Timestamp
import java.text.SimpleDateFormat
import java.util.{Date, Properties}

import breeze.numerics.log
import com.alibaba.fastjson.JSON
import redis.clients.jedis._
import com.redislabs.provider.redis
import com.central.bean._
import com.central.test.KafkaSink
import com.central.utils.RedisUtil
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.serialization.{IntegerDeserializer, StringDeserializer, StringSerializer}
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.{HashPartitioner, Partitioner, SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.json4s.NoTypeHints
import org.json4s.jackson.JsonMethods.{compact, parse, render}
import org.json4s.jackson.Serialization

import scala.util.parsing.json.JSONObject

object WarnStreaming {
  def main(args: Array[String]): Unit = {

    val duration: String = args(0)


    val conf = new SparkConf().setAppName("PoliceWarningStreaming").setMaster("local[4]")
    val ssc = new StreamingContext(conf, Seconds(duration.toInt))

    val kafkaparam = Map(
      "bootstrap.servers" -> "192.168.5.180:9092",
      "key.deserializer" -> classOf[IntegerDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "test20191106",
      "auto.offset.reset" -> "earliest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )
    val topics = Array("wifiData")

    val producer: Broadcast[KafkaSink[String, String]] = {
      val kafkaProducerConfig = {
        val p = new Properties()
        p.setProperty("bootstrap.servers", "192.168.5.180:9092")
        p.setProperty("key.serializer", classOf[StringSerializer].getName)
        p.setProperty("value.serializer", classOf[StringSerializer].getName)
        p
      }
      ssc.sparkContext.broadcast(KafkaSink(kafkaProducerConfig))
    }
    val DStream: InputDStream[ConsumerRecord[String, (String, String)]] = KafkaUtils.createDirectStream(
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, (String, String)](topics, kafkaparam)
    )
    val originData: DStream[(String, (String, Option[BehaviorDetail]), Option[OriginalDataEntity])] = DStream
      .map(x => {
        x.value()
      })
      .mapPartitions(x => x.map {
        case (idNumber, contentJson) => {
          implicit val formats = Serialization.formats(NoTypeHints)
          val standerIdNumber = if (idNumber != null) idNumber.replaceAll("[\t -]+", "").toLowerCase() else null //去掉mac或者imsi的空格和-信息
          val behaviorDetail = parse(contentJson).extractOpt[BehaviorDetail] //得到详细的轨迹信息
          if (behaviorDetail.isDefined && behaviorDetail.get.datatype.equals("105")) {
            val wifiData = parse(behaviorDetail.get.detail).extractOpt[OriginalDataEntity]

            (standerIdNumber, (behaviorDetail.get.detail, behaviorDetail), wifiData)
          } else {
            (standerIdNumber, (contentJson, None), None)
          }
        }
      })
      .persist()

    originData
      .filter(
        tuple3 => tuple3._2._2.isDefined && tuple3._2._2.get.datatype.equals("105") && tuple3._3.isDefined
      )
      .mapPartitions(
        itr =>
          itr.map { //过滤掉和增强wifi没有关系的数据并转换为WifiEntity
            case (showedIdNumber, optOriginalData, wifiDataOpt) => {
              //              import com.trs.media.utils.GlobalImplicit._
              val wifiData = wifiDataOpt.get
              if (
              //                StringUtils.isNullOrEmpty(wifiData.mac)
                wifiData == null || wifiData.mac.length == 0

              ) //imsi类型数据
                WifiEntity(wifiData.id,
                  showedIdNumber,
                  s"${wifiData.imsi}",
                  "IMSI",
                  wifiData.captime.asInstanceOf[Timestamp],
                  wifiData.captime.asInstanceOf[Timestamp],
                  wifiData.crtime.asInstanceOf[Timestamp],
                  wifiData.device_code,
                  "",
                  "",
                  "",
                  "",
                  0.00,
                  0.00,

                  "",
                  optOriginalData._1,
                  ""
                )
              else
                WifiEntity(wifiData.id,
                  showedIdNumber,
                  s"${wifiData.mac}",
                  "MAC",
                  wifiData.captime.asInstanceOf[Timestamp],
                  wifiData.captime.asInstanceOf[Timestamp],
                  wifiData.crtime.asInstanceOf[Timestamp],
                  wifiData.device_code,
                  "",
                  "",
                  "",
                  "",
                  0.00,
                  0.00,
                  "",
                  optOriginalData._1,
                  ""
                )
            }
          })

      .transform(rdd => {
        //只对中国的IMSI进行判断,主要体现在460开头
        def isStanderIMSI(imsi: String): Boolean = {
          "^460[0,1][0,1,2,3,5,6,7,8,9,4]\\d{10}$".r.findAllIn(imsi).hasNext
        }

        val config = Map("a" -> "1")
        //loadConfigFromMysql
        val errorList: List[WifiEntity] = rdd.aggregate(List.empty[(WifiEntity)])(
          (errorList, entity) => {
            val delayThreshold = config.getOrElse("validation.threshold.delay", s"${1000 * 60 * 10}").toLong
            if ((new Date().getTime - entity.behaviorStartTime.getTime) > delayThreshold) errorList :+ entity //统计数据延迟时间
            else if (!isStanderIMSI(entity.showedIdNumber)) errorList :+ entity //检查是否是标准的IMSI 去掉了entity.phoneNumber.isDefined && !
            else errorList
          },
          (errorList1, errorList2) => errorList1 ++ errorList2
        )
        //排除错误数据进行后续的计算
        val errorListRdd = rdd.sparkContext.parallelize(errorList)
        rdd.subtract(errorListRdd)
      })

      .map(
        wifiEntity => (wifiEntity.toBehaviorDetail(Some(wifiEntity.json)), wifiEntity.json)
      )
      //从redis取数据然后进行比对
      .foreachRDD(rdd => {

      //从redis读取数据
      val redisRdd: RDD[(String, String)] = RedisUtil.loadDataFromRedis(rdd, "")

      //对读取的数据进行转换
      val transRedisRdd: RDD[(String, List[FocusPersonDetail])] = RedisUtil.transRedisRdd(redisRdd)

      val filteredRdd = transRedisRdd
        .map {
          case (idNumber, contentJson) => {
            val standerIdNumber = idNumber.replaceAll("[\t -]+", "").toLowerCase() //将比对的证件号码进行规则处理
            (standerIdNumber, (idNumber, contentJson))
          }
        }
      filteredRdd
      if (filteredRdd.count() > 0) {
        //        log.infoLog("从redis加载",s"${filteredRdd.count()}条数据")

        val kafkaRdd: RDD[(String, (String, Option[BehaviorDetail]), String)] = rdd
          .map { case (entity, json) => {
            (entity.ID_NUMBER, (entity.ID_NUMBER, Option(entity)), json)
          }
          }

        //result:(ID_NUMBER,(ID_NUMBER,BehaviorDetail),Option[(ID_NUMBER,List[FocusPersonDetail])])
        val result = kafkaRdd
          .filter(item => {
            item._1 != null && item._2._2.isDefined
          })
          .map(kv => (kv._1, (kv._2._1, kv._2._2.get)))
          .leftOuterJoin(filteredRdd)
          .filter(r => r._2._2.isDefined)
        //进行排重处理
        val cleanRecords: RDD[(String, ((String, BehaviorDetail), Option[(String, List[FocusPersonDetail])]))] =
          RedisUtil.cleanTheWarnedRecords(result)(_._2._1._2.md5)
            .fold(
              rdd => rdd,
              error => {
                //            log.infoLog((new Date().getTime).toString, s"在进行排重处理时,出现错误,错误信息为:${error.getMessage}")
                ssc.sparkContext.emptyRDD[(String, ((String, BehaviorDetail), Option[(String, List[FocusPersonDetail])]))]
              }
            ).cache()
        //        log.infoLog("排重后:", s"${cleanRecords.count()}条记录被碰撞出来")
        try {
          if (cleanRecords.count() > 0) {
            personalWarning(cleanRecords, producer)
          }
        } finally {
          cleanRecords.unpersist()
        }
      }
    })
  }

  def personalWarning(joinedRdd: RDD[(String, ((String, BehaviorDetail), Option[(String, List[FocusPersonDetail])]))], producer: Broadcast[KafkaSink[String, String]]) = {

    val warningSource = joinedRdd.map(bf => {
      val showContent = bf._2 //((证件号,轨迹内容Detail),Option[(证件号,布控内容List)]
      val dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
      val dateString = dateFormat.format(new Date())
      val item = WarningMessage(
        showContent._1._1,
        showContent._1._2.toString,
        s"[${showContent._2.get._2.map(_.toString).mkString(",")}]",
        dateString,
        showContent._1._2.PRODUCER_TIME,
        "STREAMING",
        showContent._1._2.md5, "", ""
      )
      (item, showContent._1._2)
    })
    val warning = warningSource.map(_._1)

    warningSource.foreach(item => {
      val a = item._1
      val b = item._2

      val json = Map(
        "ID_NUMBER_INFO" -> a.ID_NUMBER_INFO,
        "BEHAVIOR" -> a.BEHAVIOR,
        "FOCUS_OBJECT" -> a.FOCUS_OBJECT,
        "KAFKA_TIME" -> a.KAFKA_TIME,
        "STREAMING_TIME" -> a.STREAMING_TIME,
        "ID_NUMBER" -> a.ID_NUMBER,
        "MD5" -> a.MD5,
        "ADD1FIELD" -> a.ADD1FIELD,
        "ADD2FIELD" -> a.ADD2FIELD,
        "DETAIL" -> b.detail,
        "DATATYPE" -> b.datatype
      )
      //      val json = ("ID_NUMBER_INFO" -> a.ID_NUMBER_INFO) ~
      //        ("BEHAVIOR" -> a.BEHAVIOR) ~
      //        ("FOCUS_OBJECT" -> a.FOCUS_OBJECT) ~
      //        ("KAFKA_TIME" -> a.KAFKA_TIME) ~
      //        ("STREAMING_TIME" -> a.STREAMING_TIME) ~
      //        ("ID_NUMBER" -> a.ID_NUMBER) ~
      //        ("MD5" -> a.MD5) ~
      //        ("ADD1FIELD" -> a.ADD1FIELD) ~
      //        ("ADD2FIELD" -> a.ADD2FIELD) ~
      //        ("DETAIL" -> b.detail) ~
      //        ("DATATYPE" -> b.datatype)
      producer.value.send("", JSONObject.apply(json).toString())
    })
  }
}
